/*
 * Singularity Engine
 * Copyright (C) 2010 Vyacheslav Vorobyov, All rights reserved.
 * See LICENSE for full license information.
 */

/**
 * 
 */
package org.singularity.mapred.framework;

import java.io.IOException;
import java.util.Collections;
import java.util.LinkedHashSet;
import java.util.List;

import org.singularity.io.AbstractTransitPipe;
import org.singularity.io.CloseablePipe;
import org.singularity.io.DME;
import org.singularity.mapred.RedFun;
import org.singularity.mapred.VDME;

/**
 * @author vjache
 *
 */
public class ReducerPipe<KT, RT> extends AbstractTransitPipe<VDME<KT, ?>, DME<RT>> {

	private final LinkedHashSet<Object> _mIds;

	private final RedFun<KT, RT> _redFun;

	private final ReduceCache<KT> _cache;

	public ReducerPipe(
			CloseablePipe<DME<RT>> aNextOne, 
			List<Object> aMapperIds,
			RedFun<KT, RT> aRedFun,
			ReduceCache<KT> aCache) {
		super(aNextOne);
		_mIds = new LinkedHashSet<Object>(aMapperIds);
		_redFun = aRedFun;
		_cache = aCache;
	}

	@Override
	public void push(VDME<KT, ?> aItem) throws IOException {
		if(_cache.modify(aItem))
		{
			final Object mprIdModif = aItem.getAnnotation(MapperPipe.MAPPER_ID_ANNOT);
			final Iterable<?>[] _segs = new Iterable<?>[_mIds.size()];
			int i=0;
			for(Object mprId:_mIds)
			{
				if(mprId.equals(mprIdModif))
				{
					_segs[i] = Collections.singleton(aItem.getValue()); 
				}
				else
				{
					_segs[i] = _cache.get(mprId, aItem.getKey());
				}
				
				i++;
			}
			try{
			_redFun.red(aItem.getCode(), _next, aItem.getKey(), _segs);}
			catch(Exception e)
			{
				throw new IOException(e);
			}
		}
	}
}
